package cn.murky.folkmq.utils;

import org.noear.folkmq.client.MqClient;
import org.noear.folkmq.client.MqMessage;
import org.noear.folkmq.client.MqTransaction;
import org.noear.solon.Solon;
import org.noear.solon.Utils;
import org.noear.solon.cloud.CloudClient;
import org.noear.solon.cloud.CloudProps;
import org.noear.solon.cloud.exception.CloudEventException;
import org.noear.solon.cloud.extend.folkmq.FolkmqProps;
import org.noear.solon.cloud.extend.folkmq.impl.FolkmqTransactionListener;
import org.noear.solon.cloud.model.Event;
import org.noear.solon.cloud.model.EventTran;

public class MurkyMqUtils {
    private static MqClient CLIENT;
    private static long publishTimeout;
    static {
        Solon.context().getBeanAsync(MqClient.class,bean->{
            CLIENT=bean;
        });
        Solon.context().getBeanAsync(CloudProps.class,bean->{
            publishTimeout=bean.getEventPublishTimeout();
        });
    }


    /**
     * 推送消息
     * @param event 事件模型
     * @return 推送是否成功
     */
    public static boolean publish(Event event){
        return CloudClient.event().publish(event);
    }

    /**
     * 推送消息
     * @param event 事件模型
     * @param broadcast 是否广播
     * @return 推送是否成功
     */
    public static boolean publish(Event event,boolean broadcast){
        if (Utils.isEmpty(event.topic())) {
            throw new IllegalArgumentException("Event missing topic");
        } else if (Utils.isEmpty(event.content())) {
            throw new IllegalArgumentException("Event missing content");
        } else {
            if (event.tran() != null) {
                beginTransaction(event.tran());
            }

            String topicNew = FolkmqProps.getTopicNew(event);

            try {
                MqMessage message = (new MqMessage(event.content(), event.key()))
                        .scheduled(event.scheduled())
                        .tag(event.tags())
                        .broadcast(broadcast)
                        .qos(event.qos());
                if (event.tran() != null) {
                    MqTransaction transaction = event.tran().getListener(FolkmqTransactionListener.class).getTransaction();
                    message.transaction(transaction);
                }

                if (publishTimeout > 0L) {
                    CLIENT.publish(topicNew, message);
                } else {
                    CLIENT.publishAsync(topicNew, message);
                }

                return true;
            } catch (Throwable var5) {
                throw new CloudEventException(var5);
            }
        }
    }

    private static void beginTransaction(EventTran transaction) {
        if (transaction.getListener(FolkmqTransactionListener.class) == null) {
            transaction.setListener(new FolkmqTransactionListener(CLIENT.newTransaction()));
        }
    }
}
